# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: scanners\wvs\wvs_scan_app.py
"""WVS scanning app for managing WVS scan jobs"""
import logging, os.path, time
from threading import Thread, RLock
import scanners.scan_app.scan_app_base as app_base
from scanners.scan_app.errors import *
from api.classes.scan.auto import ClsJobStatuses, ClsWorkerStatus, ClsJobStatus
from scanners.wvs.wvs_scan_job import WVSScanningJob
from server.http_responses.http_status_codes import Http200File
from settings_provider import settings
from api.classes.scan.auto import ClsJobIdentifier
from helpers.setup_types import is_setup_type_on_premise, is_setup_type_aws
from helpers.constants.jobs import ScanAppTypes
from scanners.constants import ScanJobStatus

_module_log = logging.getLogger("scan.wvs_scan_app")


class WVSScanningApp(app_base.ScanningAppBase):
    """Class taking care of multiple WVS scanning jobs"""

    def __init__(
        self, working_directory, wvs_directory=None, scan_storage_directory=None
    ):
        """Initialized the WVS app and see the capabilities of the WVS we have"""
        super().__init__()
        self.scanning_app = ScanAppTypes.WVS
        self.working_directory = working_directory
        self.scan_storage_directory = scan_storage_directory
        self.should_stop = False
        self._jobs = {}
        self._job_lock = RLock()
        self._worker_thread = Thread(target=self.worker)
        self._worker_thread.start()
        self._WVSScanningApp__wvs_directory = wvs_directory

    @property
    def wvs_directory(self):
        if self._WVSScanningApp__wvs_directory:
            if callable(self._WVSScanningApp__wvs_directory):
                return self._WVSScanningApp__wvs_directory()
            return self._WVSScanningApp__wvs_directory

    @wvs_directory.setter
    def wvs_directory(self, wvs_directory):
        self._WVSScanningApp__wvs_directory = wvs_directory

    @wvs_directory.setter
    def wvs_directory(self, new_directory):
        _module_log.info("working directory changed to %s", new_directory)
        self._WVSScanningApp__wvs_directory = new_directory

    @property
    def jobs_in_execution(self):
        jobs_in_execution = 0
        with self._job_lock:
            for job in self._jobs.values():
                if job.status not in (
                    ScanJobStatus.ABORTED,
                    ScanJobStatus.FINISHED,
                    ScanJobStatus.PAUSED,
                    ScanJobStatus.FAILED,
                ):
                    jobs_in_execution += 1

        return jobs_in_execution

    def worker(self):
        """Thread updating job statuses and doing long operations like downloads"""
        while True:
            if self.should_stop:
                _module_log.info("aborting worker process")
                for job in self._jobs.values():
                    job.abort()

                while True:
                    for job in self._jobs.values():
                        job.update_status()
                        if job.status not in (
                            ScanJobStatus.ABORTED,
                            ScanJobStatus.FINISHED,
                            ScanJobStatus.PAUSED,
                            ScanJobStatus.FAILED,
                        ):
                            break
                        else:
                            job.clean_up()
                    else:
                        return

                    time.sleep(0.5)

            try:
                delete_list = []
                if self.jobs_in_execution:
                    _module_log.debug("processing %s scans", self.jobs_in_execution)
                job_id_list = list(self._jobs.keys())
                for job_id in job_id_list:
                    job = self._jobs.get(job_id)
                    if not job:
                        continue
                    job.consume_enqueued_operations(consume_all=True)
                    status, finish_status = job.update_status()
                    _module_log.debug(
                        "update_status %s: %s, %s", job_id, status, finish_status
                    )
                    if job.can_be_deleted():
                        if job.is_marked_for_deletion:
                            delete_list.append(job_id)

                for job_id in delete_list:
                    job = self._jobs.get(job_id)
                    if job:
                        self._jobs[job_id].clean_up()

                with self._job_lock:
                    for job_id in delete_list:
                        try:
                            _module_log.debug('Deleting job "%s"', job_id)
                            del self._jobs[job_id]
                        except KeyError:
                            pass

            except Exception as e:
                _module_log.exception("Exception in main loop %s", e)

            time.sleep(1)

    @property
    def total_slots(self):
        if is_setup_type_aws():
            return settings.get("max_scans_per_engine", 4)
        elif is_setup_type_on_premise():
            from helpers.licensing.features import BaseLicense

            license_data = BaseLicense.get_license_from_license_data()
            return license_data.max_scans_per_engine
        else:
            return 0

    def _get_filtered_job_status_list(self, job_ids=None):
        """Returns a list of job statuses identified by job_ids"""
        statuses = []
        with self._job_lock:
            if job_ids is None:
                job_ids = list(self._jobs.keys())
            for job_id in job_ids:
                job = self._jobs.get(job_id)
                if job:
                    statuses.append(
                        dict(
                            job_id=job_id,
                            status=job.status,
                            extended_status=job.extended_status,
                        )
                    )
                else:
                    statuses.append(dict(job_id=job_id, status="not_found"))

        return statuses

    def shutdown(self):
        """Stops the worker thread and cleans up"""
        if not self.should_stop:
            _module_log.info("shutdown command received")
            self.should_stop = True
            self._worker_thread.join()
            _module_log.info("shutdown")

    def start_job(self, job_description):
        """Start a scanning job"""
        if self.should_stop:
            raise WorkerTerminated()
        if not self.wvs_directory:
            _module_log.error("start_job received but no wvs instance found")
            raise NoMoreWorkerSlots()
        _module_log.debug("try starting %s", job_description.job_id)
        with self._job_lock:
            job = self._jobs.get(job_description.job_id)
        if job:
            if job.can_be_deleted():
                if job.is_marked_for_deletion:
                    _module_log.warning(
                        "start_job: DuplicateJobId %s (finished previously)",
                        job_description.job_id,
                    )
                _module_log.warning(
                    "start_job: DuplicateJobId %s", job_description.job_id
                )
            raise DuplicateJobId()
        jobs_in_execution = self.jobs_in_execution
        _module_log.debug(
            "%s total slots running %s scans", self.total_slots, jobs_in_execution
        )
        if jobs_in_execution + 1 > self.total_slots:
            raise NoMoreWorkerSlots()
        try:
            new_job = WVSScanningJob(job_description, self)
        except Exception as e:
            _module_log.exception(e)
            raise StartFailed()

        with self._job_lock:
            self._jobs[job_description.job_id] = new_job
        return self.get_worker_status()

    def abort_job(self, job_identifier):
        """Abort a scanning job"""
        if self.should_stop:
            raise WorkerTerminated()
        with self._job_lock:
            try:
                status = self._jobs[job_identifier.job_id].abort()
                return ClsJobStatus(
                    job_id=job_identifier.job_id, status=status, extended_status=None
                )
            except KeyError:
                raise JobNotFound()

    def pause_job(self, job_identifier):
        """Pause a scanning job"""
        if self.should_stop:
            raise WorkerTerminated()
        with self._job_lock:
            try:
                status = self._jobs[job_identifier.job_id].pause()
                return ClsJobStatus(
                    job_id=job_identifier.job_id, status=status, extended_status=None
                )
            except KeyError:
                raise JobNotFound()

    def delete_job(self, job_identifier):
        """Delete a scanning job"""
        if self.should_stop:
            return
        with self._job_lock:
            try:
                self._jobs[job_identifier.job_id].delete_when_finished()
            except KeyError:
                return

    def get_job_status(self, job_identifier_list):
        if self.should_stop:
            raise WorkerTerminated()
        statuses = self._get_filtered_job_status_list(job_identifier_list.job_ids)
        _module_log.info("get_job_status: %s", statuses)
        return ClsJobStatuses(statuses=statuses)

    def get_worker_status(self):
        if self.should_stop:
            raise WorkerTerminated()
        return ClsWorkerStatus(
            total_slots=self.total_slots,
            empty_slots=max(0, self.total_slots - self.jobs_in_execution),
            job_statuses=self._get_filtered_job_status_list(),
        )

    def get_scan_state_db(self, scanning_app, job_id, **_):
        """
        This method is used by the backend to retrieve the scan state databases from workers
        :param scanning_app:
        :param job_id:
        :return:
        """
        if self.should_stop:
            raise WorkerTerminated()
        scan_state_storage = settings.get("storage.scan_state_db")
        if not scan_state_storage:
            _module_log.warning("scan_state_storage is not defined")
            return Http404()
        else:
            file_path = os.path.join(scan_state_storage, scanning_app, job_id)
            if not os.path.exists(file_path):
                _module_log.warning("scan state db %s is not found", file_path)
                return Http404()

            def file_reader():
                with open(file_path, "rb") as (f):
                    while True:
                        d = f.read(8192)
                        if d:
                            yield d
                        else:
                            break

            return Http200File(job_id, file_reader())
